import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.io.TimeRange;
import org.apache.hadoop.hbase.io.compress.Compression.Algorithm;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

public class Example2 implements Runnable{


    public static void main(String[] args) {
        System.out.println("Start.");

        int threadCounter=Integer.parseInt(args[0]);
        Example2 test = new Example2();


        for(int i=0;i<threadCounter;i++){
            Thread t = new Thread(test);
            t.start();
        }
    }

    @Override
    public void run() {
        Configuration config = HBaseConfiguration.create();
        try (Connection connection = ConnectionFactory.createConnection(config);
            ) {
            Table table = connection.getTable(TableName.valueOf("test"));

            for(int i=0;i<1000;i++){


                while(true){

                    Get get = new Get(Bytes.toBytes("r1"));
                    get.addColumn("cf".getBytes(),"c1".getBytes());


                    Result rr = table.get(get);
                    //long rrtimestamp = rr.getColumnCells("cf".getBytes(),"c1".getBytes()).get(0).getTimestamp();


                    int addvalue = 1;
                    int oldint = Integer.parseInt(Bytes.toString(rr.getValue("cf".getBytes(),"c1".getBytes())));
                    int newint = oldint+addvalue;

                    Put put = new Put(Bytes.toBytes("r1"));
                    put.addColumn("cf".getBytes(),"c1".getBytes(),0L,Bytes.toBytes(String.valueOf(newint)));


                    CheckAndMutateResult retry2 = table.checkAndMutate(CheckAndMutate.newBuilder(Bytes.toBytes("r1"))
                            .ifEquals("cf".getBytes(),"c1".getBytes(),Bytes.toBytes(String.valueOf(oldint)))
                            .build(put));

                    if(retry2.isSuccess()) break;
                }
            }

        }catch (Exception e){
            e.printStackTrace();
        }
    }
}